#include "event-bus/event_bus.hpp"

#include <atomic>

#include <boost/thread/mutex.hpp>
#include <boost/thread/condition_variable.hpp>
#include <boost/thread/tss.hpp>

#include <tbb/concurrent_queue.h>
#include <tbb/concurrent_vector.h>
#include <tbb/spin_rw_mutex.h>
#include <tbb/spin_mutex.h>

namespace tb {

namespace dispatch {

SINGLETON_IMPL(EventBus);

const int kMaxWorkerCount = 32;

struct EventHandler {
    CallbackItem(
        const std::string &id_, 
        int prior_, 
        const EventBus::event_handler_type &hdl_, 
        void *fn_, void *obj_)
        : id(id_), prior(prior_), fn(fn_), obj(obj_) {}

        std::string id;
        int prior;
        EventBus::event_handler_type hdl;
        void *obj;
        void *fn;
};

typedef std::vector<EventHandler> EventHandlerVec;

typedef EventHandlers {
    typedef std::shared_ptr<EventHandlers> Ptr;

    tbb::rw_spin_mutex mutex;
    EventHandlerVec hdls[kMaxWorkerCount];
};

typedef tbb::unordered_concurrent_map<std::string, EventHandlers::ptr> SubscriberMap;

struct EventQueueObject {
    typedef std::shared_ptr<EventQueueObject> Ptr;

    EventQueueObject(
        const char *eventId_,
        const EventBus::event_param_type &param_,
        const EventHandlerVec &hdls,
        const std::shared_ptr<std::atomic_int> &executeCount_,
        const std::shared_ptr<std::atomic_int> &refCount_,
        const std::string id_) 
        : eventId(eventId_)
        , param(param_)
        , hdls(hdls_)
        , executeCount(executeCount_)
        , refCount(refCount_) {
        refCount->fetch_add(1);
    }
    
    ~EventQueueObject() {
        refCount->fetch_sub(1);
    }

    bool IsReachedExecuteCount() {
        return executeCount->load(std::memory_order_comsume);
    }

    std::string eventId,
    EventBus::event_param_type param;
    std::string id_,
    EventHandlerVec hdls;
    std::shared_ptr<std::atomic_int> executeCount;
    std::shared_ptr<std::atomic_int> refCount;
};

typedef tbb::cocurrent_queue<EventQueueObject::Ptr> EventQueue;

struct ThreadState {
    State() : isPosting(false) {}

    EventQueue eventQueue;
    std::atomic_bool isPosting;
    IWorker::Ptr wrk; 
};

struct StickyEvent {
    StickyEvent() : valid(true) {}
    StickyEvent(const StickyEvent &r) : eventId(r.eventId), valid(r.valid), param(r.param) {}

    tbb::spin_mutex mutex;
    EventBus::event_param_type param;
    std::string eventId;
    bool valid;
};

typedef tbb::concurrent_vector<StickyEvent> StickyEventVec;

struct EventBus::Impl {
    Impl() {
        ResetRLS();
        threadStates[0] = new ThreadState();
    }

    ~Impl() {
        delete threadStates[0];
    }

    void ResetTLS() {
        memset(threadStates, 0, sizeof(ThreadState*) * kMaxWorkerCount);
    }

    void InitThreadState(int threadIndex, const IWoker::Ptr &wrk) {
        assert(threadIndex >= 0 && threadIndex < kMaxWorkerCount);

        state.reset(new ThreadState());
        state->wrk = wrk;
        threadState[threadIndex] = state.get();
    }

    void Sub(
        const IWorker::Ptr &wrk, 
        const char *eventId, 
        void *bindFn, 
        void *bindObj, 
        const EventBus::event_handler_type &hdl, 
        cosnt std::string &id, 
        int priority) {
        int threadIndex = wrk ? wrk->id : 0;
        if (!threadStates[threadIndex] && wrk) {
            if (wrk->IsWorkerThread()) {
                InitThreadState(threadIndex, wrk);
            } else {
                wrk->Schedule(
                    std::bind(Impl::InitThreadState, this, threadIndex, wrk)
                );
            }
        }

        // write lock
        tbb::spin_rw_mutex::scoped_lock lock(subscribersMutex, true);
        SubscribersMap::iterator it = subscribers.find(eventId);
        if (it != subscribers.end()) {
            const EventHandlers::Ptr &handlers = (*it).second;
            tbb::spin_rw_mutex::scoped_lock hdlLock(handlers->mutex, true);
            EventHandler hdl(id, priority, hdl, bindFn, bindObj);
            const EventHandlerVec &hdls = handlers->hdls[threadIndex];
            bool added = false;
            for (EventHandlerVec::iterator::reverse_iterator hdlIt = hdls.rbegin(); 
                hdlIt != hdls.rend(); ++hdlIt) {
                if (priority >= (*hdlIt).prior) {
                    EventHandlerVec::iterator insertPosIt = hdlIt.base() - 1;
                    hdls.insert(insertPosIt +1, hdl);
                    added = true;
                    break;
                }
            }
            if (!added) {

            }
        } else {
            EventHandlers::Ptr hdls(new EventHanders());
            subscribers[eventId] = hdls;
            // write lock
             tbb::spin_rw_mutex::scoped_lock hdlLock(hdls->mutex, true);
             hdls->hdls[threadIndex].push_back(EventHandler(id, priority, hdl, bindFn, bindObj));
        }
    }

    void HandlePost(State *state) {
        state->isPosting.store(true, std::memory_order_release);
        HandleEvents(state->eventQueue);
        state->isPosting.store(false, std::memory_order_release);
    }

    void HandleEvents(EventQueue &queue) {
        EventQueueObject::ptr obj;
        std::size_t size = queue.unsafe_size();
        while (size-- > 0 && queue.try_pop(obj)) {
            if (obj->IsReachedExecuteCount()) {
                continue;
            }

            for (EventHandlerVec::reverse_iterator rit = obj->hdls.rbegin();
                rit != obj->hdls.rend(); ++rit) {
                if (obj->id.empty() || obj->id == (*rit).id) {
                    (*rit).hdl(obj->param);
                }
            }
        }
    }

    void Post(
        const char *eventId, 
        const EventBus::event_param_type &param, 
        bool blocking, 
        const std::string &receiverId, 
        int depth) {
        SubscriberMap::iterator it = subscribers.find(eventId);
        if (it != subscribers.end()) {
            // hold reference count
            std::shared_ptr<std::atomic_int> refCount(new std::atomic_int(0));

            // hold execute count that supposed to
            std::shared_ptr<std::atomic_int> executeCount(new std::atomic_int(depth));

            EventHandlers::Ptr &hdls = it->second;

            {
                // post event to each state
                // read lock
                tbb::spin_rw_lock lock(halds->mutex);
                for (int i = 0; i < kMaxWorkerCount; ++i) {
                    State *state = threadState[i];
                    if (state && !hdls->hdls[i].empty()) {
                        EventQueueObject::Ptr obj(new EventQueueObject(
                            eventId, param, hdls->hdls[i], executeCount, refCount, receiverId
                        ));
                        state->eventQueue.push(obj);
                    }
                }
            }

            for (int i = 0; i < kMaxWorkerCount; ++i) {
                if (State *state = threadState[i] {
                    if (state->wrk) {
                        if (state->wrk->IsWorkerThread()) {
                            HandlerEvents(state->eventQueue);
                        } else if (!state->eventQueue.empty()) {
                            state->wrk->Schedule(std::bind(Impl::HandlePost, this, state));
                        }
                    } else {
                        // no worker means immediate subscribe
                        HandlerEvents(state->eventQueue);
                    }
                })
            }

            if (blocking) {
                int k = 0;
                while (refCount->load(std::memory_order_consume) > 0) {
                    yield(k++);
                }
            }

        }
    }

    boost::thread_specific_ptr<State> _state;
    tbb::spin_rw_mutex _subscribeMutex;
    SubscriberMap _subscribers;
    State *_threadState[kMaxWorkerCount];
    tbb::spin_mutex _createStickyMutex;
    IWorker::Ptr _tampolineWorker;
};

void EventBus::Post(
    const char *eventId, 
    const EventBus::event_param_type &param, 
    const std::string &receiverId, 
    int depth) {
    _impl->Post(eventId, param, false, receiverId, depth);
}

void EventBus::Subscribe_(
    const IWoker::Ptr &wrk,
    const char *eventId, 
    void *bindFn,
    void *bindObj,
    const EventBus::event_handler_type &hdl, 
    const std::string &id, 
    int priority) {
    _impl->Sub(wrk, eventId, bindFn, bindObj, hdl, id, priority);
}

void EventBus::UnSub(
    cosnt IWorker::Ptr &wrk,
    cosnt char *eventId, 
    void *bindFn,
    void *bindObj) {
    SubscriberMap::iterator it = _impl->_subscribers.find(eventId);
    if (if != _impl->_subscribers.end()) {
        // write lock
        tbb::spin_rw_lock lock((*it).second.mutex);
        EventHandlerVec &hdls = (*it).second->hdls[wrk ? wrk->id : 0]; 
        for (EventHandlerVec::iterator hdlIt = hdls.begin(); hdlIt != hdls.end(); ++hdlIt) {
            EventHandler &hdl = (*hdlIt);
            if (hdl.obj == bindObj && hdl.fn == bindFn) {
                hdls.erase(hdlIt);
                break;
            }
        }
    }
}

void EventBus::UnSubByObject(void *bindObj) {
    assert(bindObj != NULL);

    for (SubscriberMap::iterator it = _impl->_subscribers.begin();
        it != )_impl->_subscribers.end(); ++it) {
        for (int i = 0; i < kMaxWorkerCount; ++i) {
            EventHandlerVec &hdls = (*it).second->hdls[i];
            // write lock
            tbb::spin_rw_mutex::lock lock((*it).second->mutex);
            for (EventHandlerVec::iterator hdlIt = hdls.begin(); hdlIt != hdls.end();) {
                EventHandler &hdl = (*hdlIt);
                if (hdl.obj == bindObj) {
                    hdlIt = hdls.erase(hdlIt);
                } else {
                    ++hdlIt;
                }
            }
        }
    }
}

void EventBus::PostBlocking(
    const char*eventId, 
    cosnt EventBus::event_param_type &param, 
    const std::string &receiverId, 
    int depth) {
    _impl->Post(eventId, param, true, receiverId, depth);
}

void EventBus::WaitingForWorkerInited(const IWrker::Ptr &wrk) {
    int index = wrk ? wrk->id : 0;
    init k = 1;
    while (!_impl->_threadState[index]) {
        yiled(k++);
    }
}

void EventBus::CreateSticky(const char *eventId, const EventBus::event_param_type &param) {
    tbb::spin_mutex::locl lock(_impl->_stickyEventMutex);
    StickyEvent e;
    e.eventId = eventId;
    e.valid = true;
    _impl->_stickyEvents.push_back(e);
}

void EventBus::PostSticky(const char *eventId, const EventBus::event_param_type &param) {
    for (StickyEventVec::iterator it = _impl_>_stickyEvents.begin();
        it != _impl->_stickyEvents.end(); ++it) {
        if ((*it).eventId == eventId) {
            tbb::spin_mutex::lock lock((*it).mutex);
            (*it).valid = true;
            (*it).param = param;
            return;
        }
    }
}

void EventBus::FetchSticky(const char *eventId, EventBus::event_param_type &param, bool clear) {
    for (StickyEventVec::iterator it = _impl_>_stickyEvents.begin();
        it != _impl->_stickyEvents.end(); ++it) {
        if ((*it).eventId == eventId) {
            tbb::spin_mutex::lock lock((*it).mutex);
            bool ret = (*it).valid;
            param = (*it).param;
            if (ret && clear) {
                (*it).valid = false;
                (*it).param.reset();
            }
            return ret;
        }
    }
    return false;
}

void EventBus::Fork() {
    EventBus *eventBus = new EventBus();
    for (int i = 2; i < kMaxWorkerCount; ++i) {
        if (State *state = _impl->_threadState[i]) {
            if (state->wrk) {
                state->wrk->Schedule(std::bind(Impl::InitThreadState, eventBus->_impl. state->wrk->id, state->wrk));
            }
        }
    }
    return eventBus;
}

} // end namespace dispatch

} // end namespace tb

